package com.client;

import com.Async.RpcFuture;
import com.serializetool.Request;
import com.serializetool.handler.RpcDecoder;
import com.serializetool.handler.RpcEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;


public class ClientBuilder {

    private Bootstrap bootStrap;
    private EventLoopGroup workerGroup;
    private ConcurrentHashMap channelMap;
    private ConcurrentHashMap<Long,RpcFuture> responseListener;
    private ConcurrentHashMap<Long,Request> requestStorage;
    private static final Object PRESENT = new Object();


    public ClientBuilder(){
        channelMap = new ConcurrentHashMap<>();
        responseListener = new ConcurrentHashMap<>();
        workerGroup = new NioEventLoopGroup();
        try {
            bootStrap = new Bootstrap();
            ClientHandler handler = new ClientHandler(this);
            bootStrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new RpcEncoder());
                            ch.pipeline().addLast(new RpcDecoder());
                            ch.pipeline().addLast(handler);
                        }
                    })
                    .option(ChannelOption.SO_KEEPALIVE, true);;
            System.out.println("正在调用客户端");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    public ChannelFuture connect(InetSocketAddress address) throws InterruptedException {
        ChannelFuture channelFuture = null;
        channelFuture = bootStrap.connect(address.getAddress(),address.getPort()).sync();
        return channelFuture;
    }

    public ConcurrentHashMap<Long, RpcFuture> getResponseListener() {
        return responseListener;
    }

    public void send(InetSocketAddress socket, Request request, RpcFuture future) throws InterruptedException {
        String strSocket = socket.toString();
        Channel channel = null;
        Object o = channelMap.putIfAbsent(strSocket,PRESENT);
        if(o==null){
            ChannelFuture channelFuture = connect(socket);
            channel = channelFuture.channel();
            channelMap.put(strSocket,channel);
        }
        while((o=channelMap.get(strSocket))==PRESENT);
        channel = (Channel) o;
        responseListener.put(request.getTaskId(),future);
        requestStorage.put(request.getTaskId(),request);
        channel.writeAndFlush(request);
    }

    public ConcurrentHashMap<Long, Request> getRequestStorage() {
        return requestStorage;
    }

    public ConcurrentHashMap getChannelMap() {
        return channelMap;
    }

    public void close(){
        if(bootStrap!=null){
            return;
        }
        workerGroup.shutdownGracefully();
    }
}
